package me.huha.service.impl;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import me.huha.client.ElasticSearchClient;
import me.huha.commons.ElasticSearchDTO;
import me.huha.service.ElasticSearchService;
import me.huha.utils.JsonUtils;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;

/**
 * @Title: ElasticSearchServiceImpl.java
 * @Author: Yepx
 * @mail: yepx2013@gmail.com
 * @CreateTime: 2017年9月29日 下午3:33:22
 * @Todo: TODO
 */
public class ElasticSearchServiceImpl implements ElasticSearchService {
	
	private static TransportClient client;
	
	static{
		try {
			client = ElasticSearchClient.getInstance();
		} catch (UnknownHostException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	@Override
	public JSONObject create(String indices, String types, JSONArray data){
		
		indices = indices.toLowerCase();
		types = types.toLowerCase();
		
		JSONObject result = new JSONObject();
		if(data == null || data.size() <= 0){
			
			result.put("code", 403);
			result.put("msg", "data 数据缺失");
			return result;
		}else{
			if(client != null){
				List<JSONObject> errorData = new ArrayList<>();
				for (Object object : data) {
					
					XContentBuilder xContentBuilder = JsonUtils.model2Json(JSONObject.fromObject(object));
					IndexResponse response = client.prepareIndex(indices, types).setSource(xContentBuilder).get();
					
					if(response.status().getStatus() == 201){
	                	System.out.println("请求成功并且服务器创建了新的资源。");
	                }else{
	                	
	                	JSONObject jsonObj = JSONObject.fromObject(object);
	                	errorData.add(jsonObj);
	                }
				}
				
				if (errorData.size() == data.size()) {
					result.put("code", 501);
					result.put("msg", "插入失败");
				}else{
					result.put("code", 200);
					if(errorData.size() == 0){
						result.put("msg", "请求成功并且服务器创建了新的资源");
					}else{
						result.put("msg", "请求成功并且服务器创建了部分新的资源");
						result.put("errorData", errorData);
					}
					
				}
				
			}else{
				result.put("code", 500);
				result.put("msg", "插入失败，ElasticSearchClient初始化异常");
			}
		}
		return result;
	}
	@Override
	public JSONObject query(String indices, String types, Map<String, Object> searchKey) {
		
		indices = indices.toLowerCase();
		types = types.toLowerCase();
		
		JSONObject result = new JSONObject();
		
		Set<String> keySet = new HashSet<String>();
		if(searchKey != null){
			keySet = searchKey.keySet();
		}
		
		if(client != null){
			
			SearchRequestBuilder searchBuider = client.prepareSearch(indices).setTypes(types);
			if(keySet != null && keySet.size() > 0){
				for (String key : keySet) {
					QueryBuilder qb= QueryBuilders.multiMatchQuery(searchKey.get(key), key);
					searchBuider = searchBuider.setQuery(qb);
				}
			}else{
				searchBuider = searchBuider.setQuery(null);
			}
			 SearchHits hits = searchBuider.execute().actionGet().getHits();
			 
			 List<ElasticSearchDTO> list = new ArrayList<>();
			 
			 if(hits.getTotalHits() > 0){
				 for (SearchHit hit : hits) {
					String id = hit.getId();
					String index = hit.getIndex();
					String type = hit.getType();
					Map<String, Object> source = hit.getSource();
					
					ElasticSearchDTO dto = new ElasticSearchDTO(id, index, type, source);
					list.add(dto);
				}
			 }else{
				 System.out.println("搜到0条结果");
			 }
			 result.put("code", 200);
			 result.put("msg", "查询成功");
			 result.put("data", list);
		}else{
			result.put("code", 500);
			result.put("msg", "查询失败，ElasticSearchClient初始化异常");
		}
		return result;
	}
	
	@Override
	public JSONObject deleteIndex(String indices){
		
		indices = indices.toLowerCase();
		
		JSONObject result = new JSONObject();
		if(client != null){
			DeleteIndexResponse dResponse = client.admin().indices().prepareDelete(indices).execute().actionGet();
			if (dResponse.isAcknowledged()) {
                System.out.println("delete index "+indices+" successfully!");
                result.put("code", 200);
                result.put("msg", "索引库：\""+ indices + "\",删除成功");
            }else{
                System.out.println("Fail to delete index " + indices);
                result.put("code", 501);
                result.put("msg", "索引库：\""+ indices + "\",删除失败");
            }
		}else{
			result.put("code", 500);
			result.put("msg", "删除失败，ElasticSearchClient 初始化异常");
		}
		return result;
	}
	
	@Override
	public JSONObject deleteIndexBykeyAndValue(String indices, String types, String id){
		
		indices = indices.toLowerCase();
		types = types.toLowerCase();
		
		JSONObject result = new JSONObject();
		if(client != null){
			DeleteResponse dResponse = client.prepareDelete().setIndex(indices).setType(types).setId(id).execute().actionGet();
			if (dResponse.status().getStatus() == 200) {
                System.out.println("delete index "+indices+" - "+types+" - " + id + " successfully!");
                result.put("code", 200);
                result.put("msg", "索引库：\""+ indices + "\"内的 \"" + types+ "\" 中的 \""+id+ "\",删除成功");
            }else{
                System.out.println("Fail to delete index " + indices+" - "+types+" - "+id);
                
                result.put("code", 501);
                result.put("msg", "索引库：\""+ indices + "\"内的 \"" + types+ "\" 中的 \""+id+ "\",删除失败");
            }
		}else{
			result.put("code", 500);
			result.put("msg", "删除失败，ElasticSearchClient 初始化异常");
		}
		return result;
	}
	
	@Override
	public JSONObject update(String indices, String types, String id, Map<String, Object> data){
		
		indices = indices.toLowerCase();
		types = types.toLowerCase();
		
		JSONObject result = new JSONObject();
		if(data == null || data.isEmpty()){
			result.put("code", 403);
			result.put("msg", "data 数据为空");
			return result;
		}
		if(client != null){
			//TODO 创建一个UpdateRequest,然后将其发送给client.
			try {
				Set<String> keySet = data.keySet();
				//TODO 创建 XContentBuilder
				XContentBuilder xBuilder = jsonBuilder().startObject();
				for (String key : keySet) {
					xBuilder = xBuilder.field(key, data.get(key));
				}
				xBuilder.endObject();
				
				UpdateRequest uRequest = new UpdateRequest();
				uRequest.index(indices);
	            uRequest.type(types);
	            uRequest.id(id);
	            uRequest.doc(xBuilder);
	            
	            UpdateResponse uResponse = client.update(uRequest).get();
	            System.out.println(uResponse.status().getStatus());
	            if(uResponse.status().getStatus() == 201 || uResponse.status().getStatus() == 200){
	             	System.out.println("更新成功");
	             	
	             	result.put("code", 200);
					result.put("msg", "更新成功");
	            }else{
	             	System.out.println("更新失败");
	             	
	             	result.put("code", 500);
					result.put("msg", "更新失败");
	            }
	            
			} catch (Exception e) {
				result.put("code", 500);
				result.put("msg", "更新失败");
			}
            
		}else{
			result.put("code", 500);
			result.put("msg", "更新失败，ElasticSearchClient 初始化异常");
		}
		return result;
	}
	@Override
	public JSONObject createOrUpdate(String indices, String types, String id, Map<String, Object> oldData, Map<String, Object> newData){
		
		indices = indices.toLowerCase();
		types = types.toLowerCase();
		
		JSONObject result = new JSONObject();
		if(oldData == null || oldData.isEmpty()){
			result.put("code", 403);
			result.put("msg", "data 数据为空");
			return result;
		}
		if(newData == null || newData.isEmpty()){
			result.put("code", 403);
			result.put("msg", "data 数据为空");
			return result;
		}
		if(client != null){
			//TODO upsert 如果文档不存在则创建新的索引
			Set<String>	oldkeySet = oldData.keySet();
			Set<String> newKeySet = newData.keySet();
			try{
				//TODO 构造查询数据
				XContentBuilder oldXbuider = jsonBuilder().startObject();
				for (String key : oldkeySet) {
					oldXbuider = oldXbuider.field(key, oldData.get(key));
				}
				oldXbuider.endObject();
				IndexRequest indexRequest = new IndexRequest(indices, types, id).source(oldXbuider);
				
				//TODO 更新数据
				XContentBuilder newXbuider = jsonBuilder().startObject();
				for (String key : newKeySet) {
					newXbuider = newXbuider.field(key, newData.get(key));
				}
				newXbuider.endObject();
				
				UpdateRequest updateRequest = new UpdateRequest(indices, types, id).doc(newXbuider).upsert(indexRequest);
				UpdateResponse updateResponse = client.update(updateRequest).get();
				
				System.out.println(updateResponse.status().getStatus());
				
				if(updateResponse.status().getStatus() == 201 || updateResponse.status().getStatus() == 200){
	             	System.out.println("更新成功");
	             	
	             	result.put("code", 200);
					result.put("msg", "更新成功");
	            }else{
	             	System.out.println("更新失败");
	             	
	             	result.put("code", 500);
					result.put("msg", "更新失败");
	            }
				 
			}catch(Exception e){
				result.put("code", 500);
				result.put("msg", "更新失败");
			}
			
		}else{
			result.put("code", 500);
			result.put("msg", "更新失败，ElasticSearchClient 初始化异常");
		}
		return result;
	}
	
	@Override
	public JSONObject batchBulkIn(String indices, String types, String filePath){
		
		indices = indices.toLowerCase();
		types = types.toLowerCase();
		
		JSONObject result = new JSONObject();
		if(client == null){
			result.put("code", 500);
			result.put("msg", "查询失败，ElasticSearchClient初始化异常");
			
			return result;
		}

		try {

			 File article = new File(filePath);
	         FileReader fr = new FileReader(article);
	         BufferedReader bfr = new BufferedReader(fr);
	         String line = null;
	         BulkRequestBuilder bulkRequest = client.prepareBulk();
	         int count = 0;
	         
	         while((line=bfr.readLine()) != null){
	        	 bulkRequest.add(client.prepareIndex(indices,types).setSource(line));
	        	 if (count%10 == 0) {
	                 bulkRequest.execute().actionGet();
	             }
	             count++;
	             System.out.println(line);
	         }
	         
	         bulkRequest.execute().actionGet();

	         bfr.close();
	         fr.close();
	         
	         result.put("code", 200);
			 result.put("msg", "导入成功");
		} catch (Exception e) {
			// TODO: handle exception
			result.put("code", 500);
			 result.put("msg", "导入失败");
		}
		
		return result;
	}
	
	@Override
	public JSONObject batchBulkOut(String indices, String types, String filePath){
		
		indices = indices.toLowerCase();
		types = types.toLowerCase();
		
		JSONObject result = new JSONObject();
		if(client == null){
			result.put("code", 500);
			result.put("msg", "查询失败，ElasticSearchClient初始化异常");
			
			return result;
		}
		try {
			SearchResponse response = client.prepareSearch(indices)
							                .setTypes(types)
							                .setQuery(QueryBuilders.matchAllQuery())
							                .execute().actionGet();
			
			SearchHits resultHits = response.getHits();
			File article = new File(filePath);
			FileWriter fw = new FileWriter(article);
			BufferedWriter bfw = new BufferedWriter(fw);
			
			if (resultHits.getHits().length == 0) {
	             System.out.println("查到0条数据!");
	         } else {
	             for (int i = 0; i < resultHits.getHits().length; i++) {
	                 String jsonStr = resultHits.getHits()[i]
	                         .getSourceAsString();
	                 System.out.println(jsonStr);
	                 bfw.write(jsonStr);
	                 bfw.write("\n");
	             }
	         }
	         bfw.close();
	         fw.close();
	         
	         result.put("code", 200);
	         result.put("msg", "导出成功");
		} catch (Exception e) {
			// TODO: handle exception
			result.put("code", 500);
	        result.put("msg", "导出失败");
		}
		
		return result;
	}
}
